package com.angle.im.server.config.mq;

import com.aliyun.openservices.ons.api.*;
import com.angle.im.server.config.mq.listeners.SendListListener;
import com.angle.im.server.config.mq.listeners.SendOneListener;
import com.angle.im.server.domain.constants.MqTags;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * @author zhangwei
 * @date 2018/9/29 18:51
 * @desc MQ消费者配置信息
 */
@SuppressWarnings("ALL")
@Configuration
@Slf4j
public class ConsumerConfiguration {

    private final String ACCESS_KEY = "your ons access key";
    private final String SECRET_KEY = "your ons secret key";
    private final String TOPIC = "demo-topic";
    private final String SEND_TO_ONE = "GID_SEND_TO_ONE";
    private final String SEND_TO_LIST = "GID_SEND_TO_LIST";

    @Autowired
    private SendOneListener sendOneListener;
    @Autowired
    private SendListListener sendListListener;

    /**
     * 对用户列表发送消息的监听
     *
     * @return
     */
    @Bean
    public Consumer sendListListenerConsumer() {
        return start(SEND_TO_LIST,sendListListener, MqTags.SEND_TO_LIST);
    }

    /**
     * 对指定用户发送消息的监听
     *
     * @return
     */
    @Bean
    public Consumer sendOneListenerConsumer() {
        return start(SEND_TO_ONE,sendOneListener, MqTags.SEND_TO_ONE);
    }


    /**
     * 开启消息监听
     *
     * @param listener 监听器,自定义实现
     * @param tags     消息监听标签
     * @return 消费者
     */
    private Consumer start(final String groupId,MessageListener listener, String... tags) {
        StringBuilder tag = new StringBuilder();
        if (tags == null || tags.length == 0) {
            throw new RuntimeException("消息tag不能为空");
        } else if (tags.length != 1) {
            for (String s : tags) {
                tag.append(s).append("||");
            }
            tag = new StringBuilder(tag.substring(0, tag.lastIndexOf("||")));
        } else {
            tag = new StringBuilder(tags[0]);
        }
        Consumer consumer = getBroadCastingConsumer(groupId);
        consumer.subscribe(TOPIC, tag.toString(), listener);
        consumer.start();
        log.info("{} started",groupId);
        return consumer;
    }

    /**
     * 获取广播模式的mq消费者
     *
     * @param consumerId
     * @return
     */
    private Consumer getBroadCastingConsumer(final String groupId) {
        Properties properties = getMqCommonProperties();
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        // 订阅方式
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        return ONSFactory.createConsumer(properties);
    }

    /**
     * 获取集群订阅模式的消费者
     *
     * @param consumerId
     * @return
     */
    private Consumer getClusteringConsumer(final String groupId) {
        Properties properties = getMqCommonProperties();
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        // 订阅方式
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        return ONSFactory.createConsumer(properties);
    }

    private Properties getMqCommonProperties() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.AccessKey, ACCESS_KEY);
        properties.put(PropertyKeyConst.SecretKey, SECRET_KEY);
        properties.put(PropertyKeyConst.MaxReconsumeTimes, "5");
        // 设置 TCP 接入域名（公共云生产环境）
        String onsAddr = "http://onsaddr.mq-internet-access.mq-internet.aliyuncs.com:80";
        properties.put(PropertyKeyConst.NAMESRV_ADDR, onsAddr);
        return properties;
    }
}
